package org.go.workstore.memory;

import java.util.ArrayList;
import java.util.Comparator;
import java.util.Date;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.TreeSet;
import java.util.concurrent.atomic.AtomicLong;

import org.go.domain.CompletedExecutionInstruction;
import org.go.domain.TriggerType;
import org.go.expcetion.GoException;
import org.go.expcetion.JobPersistenceException;
import org.go.expcetion.ObjectAlreadyExistsException;
import org.go.scheduler.SchedulerSignaler;
import org.go.spi.TriggerFiredBundle;
import org.go.spi.TriggerFiredResult;
import org.go.trigger.Calendar;
import org.go.trigger.Trigger;
import org.go.trigger.Trigger.TriggerTimeComparator;
import org.go.trigger.TriggerKey;
import org.go.trigger.TriggerWrapper;
import org.go.util.Key;
import org.go.work.Work;
import org.go.work.WorkKey;
import org.go.work.WorkWrapper;
import org.go.workstore.WorkStore;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * 
 * @author hejie
 *
 */
public class WorkStoreMemory implements WorkStore {
	class TriggerWrapperComparator implements Comparator<TriggerWrapper> {

		TriggerTimeComparator ttc = new TriggerTimeComparator();

		@Override
		public int compare(TriggerWrapper trig1, TriggerWrapper trig2) {
			return ttc.compare(trig1.trigger, trig2.trigger);
		}

		@Override
		public boolean equals(Object obj) {
			return (obj instanceof TriggerWrapperComparator);
		}

		@Override
		public int hashCode() {
			return super.hashCode();
		}
	}

	private static final AtomicLong ftrCtr = new AtomicLong(System.currentTimeMillis());

	private Logger log = LoggerFactory.getLogger(WorkStoreMemory.class);
	//private List<Map<String, Object>> workers = new ArrayList<Map<String, Object>>();
	protected HashSet<WorkKey> blockedJobs = new HashSet<WorkKey>();
	protected HashMap<String, Calendar> calendarsByName = new HashMap<String, Calendar>(25);
	protected HashMap<WorkKey, WorkWrapper> jobsByKey = new HashMap<WorkKey, WorkWrapper>(1000);
	protected final Object lock = new Object();
	protected long misfireThreshold = 5000L;
	protected HashSet<String> pausedJobGroups = new HashSet<String>();
	protected HashSet<String> pausedTriggerGroups = new HashSet<String>();
	protected SchedulerSignaler signaler;
	protected TreeSet<TriggerWrapper> timeTriggers = new TreeSet<TriggerWrapper>(new TriggerWrapperComparator());
	protected ArrayList<TriggerWrapper> triggers = new ArrayList<TriggerWrapper>(1000);
	protected HashMap<String, HashMap<TriggerKey, TriggerWrapper>> triggersByGroup = new HashMap<String, HashMap<TriggerKey, TriggerWrapper>>(25);
	protected HashMap<TriggerKey, TriggerWrapper> triggersByKey = new HashMap<TriggerKey, TriggerWrapper>(1000);

	private long getMisfireThreshold() {
		return misfireThreshold;
	}

	private boolean removeTrigger(TriggerKey key, boolean removeOrphanedJob) {
		synchronized (lock) {

		}
		return true;
	}

	private String ttt(String id) {
		if (id == null)
			return "";
		for (TriggerWrapper tw : triggers) {
			String key = tw.getTrigger().getTriggerKey();
			if (id.equals(key)) {
				return tw.getTrigger().getType();
			}
		}
		return "";
	}

	protected boolean applyMisfire(TriggerWrapper tw) {

		long misfireTime = System.currentTimeMillis();
		if (getMisfireThreshold() > 0) {
			misfireTime -= getMisfireThreshold();
		}

		Date tnft = tw.trigger.getNextFireTime();
		if (tnft == null || tnft.getTime() > misfireTime || tw.trigger.getMisfireInstruction() == Trigger.MISFIRE_INSTRUCTION_IGNORE_MISFIRE_POLICY) {
			return false;
		}

		Calendar cal = null;
		if (tw.trigger.getCalendarName() != null) {
			cal = retrieveCalendar(tw.trigger.getCalendarName());
		}

		signaler.notifyTriggerListenersMisfired(tw.trigger.clone());

		tw.trigger.updateAfterMisfire(cal);

		if (tw.trigger.getNextFireTime() == null) {
			tw.state = TriggerWrapper.STATE_COMPLETE;
			signaler.notifySchedulerListenersFinalized(tw.trigger);
			synchronized (lock) {
				timeTriggers.remove(tw);
			}
		} else if (tnft.equals(tw.trigger.getNextFireTime())) {
			return false;
		}

		return true;
	}

	protected String getFiredTriggerRecordId() {
		return String.valueOf(ftrCtr.incrementAndGet());
	}

	/*
	 * (non-Javadoc)
	 * @see org.go.core.WorkerStore#acquireNextTriggers(long, int, long)
	 */
	@Override
	public List<Trigger> acquireNextTriggers(long noLaterThan, int maxCount, long timeWindow) throws GoException {
		synchronized (lock) {
			log.debug("triggers.size(): {}", triggers.size());
			log.debug("triggers.size(): {}", timeTriggers.size());
			List<Trigger> result = new ArrayList<Trigger>();
			while (true) {
				TriggerWrapper tw;
				try {
					tw = timeTriggers.first();
					if (tw == null) {
						return result;
					}
					timeTriggers.remove(tw);
				} catch (java.util.NoSuchElementException nsee) {
					return result;
				}

				if (tw.trigger.getNextFireTime() == null) {
					continue;
				}

				if (applyMisfire(tw)) {
					if (tw.trigger.getNextFireTime() != null) {
						timeTriggers.add(tw);
					}
					continue;
				}

				if (tw.getTrigger().getNextFireTime().getTime() > noLaterThan + timeWindow) {
					timeTriggers.add(tw);
					return result;
				}
				tw.state = TriggerWrapper.STATE_ACQUIRED;
				tw.trigger.setFireInstanceId(getFiredTriggerRecordId());
				Trigger trig = tw.trigger.clone();
				result.add(trig);
				//
				if (result.size() == maxCount) {
					return result;
				}
			}
		}
	}

	/*
	 * (non-Javadoc)
	 * @see org.go.core.WorkerStore#addWorkDesc(org.go.WorkerDesc)
	 */
	@Override
	public void addWorkDesc(Work workerDesc) {
		WorkWrapper ww = new WorkWrapper(workerDesc);
		jobsByKey.put(new WorkKey(Key.createUniqueName(null)), ww);
	}

	/*
	 * (non-Javadoc)
	 * @see org.go.core.WorkerStore#queryWorkDesc(java.lang.String)
	 */
	@Override
	public Work queryWorkDesc(String id) {
		//		for (Map<String, Object> map : workers) {
		//			String tmpId = (String) map.get("KEY");
		//			if (tmpId.equals(id)) {
		//				return (WorkerDesc) map.get("VALUE");
		//			}
		//		}
		return null;
	}

	/*
	 * (non-Javadoc)
	 * @see org.go.core.WorkerStore#queryWorkDesc(org.go.domain.TriggerType)
	 */
	@Override
	public Work[] queryWorkDesc(TriggerType type) {
		//		List list = new ArrayList();
		//		Iterator<Entry<WorkKey, WorkerWrapper>> iterator = jobsByKey.entrySet().iterator();
		//		while (iterator.hasNext()) {
		//			Entry<WorkKey, WorkerWrapper> entry = iterator.next();
		//			WorkerWrapper workerWrapper = entry.getValue();
		//			String id = workerWrapper.getWorkerDesc().getTriggerId();
		//			String ttt = ttt(id);
		//			if (type.getName().equals(ttt)) {
		//				list.add(workerWrapper.getWorkerDesc());
		//			}
		//		}
		//		Work[] ks = (Work[]) list.toArray(new Work[] {});
		//		return ks;
		return null;
	}

	/*
	 * (non-Javadoc)
	 * @see org.go.core.WorkerStore#releaseAcquiredTrigger(org.go.Trigger)
	 */
	@Override
	public void releaseAcquiredTrigger(Trigger trigger) throws GoException {

	}

	/*
	 * (non-Javadoc)
	 * @see org.go.core.WorkerStore#retrieveCalendar(java.lang.String)
	 */
	@Override
	public Calendar retrieveCalendar(String calName) {
		synchronized (lock) {
			Calendar cal = calendarsByName.get(calName);
			if (cal != null)
				return (Calendar) cal.clone();
			return null;
		}
	}

	/*
	 * (non-Javadoc)
	 * @see org.go.core.WorkerStore#retrieveJob(org.go.WorkKey)
	 */
	@Override
	public Work retrieveJob(WorkKey jobKey) {
		synchronized (lock) {
			if (jobKey == null) {
				jobKey = new WorkKey(Key.createUniqueName(null));
			}
			WorkWrapper jw = jobsByKey.get(jobKey);
			//return (jw != null) ? (Work) jw.work.clone() : null;
			return (jw != null) ? (Work) jw.work : null;
		}
	}

	public void setMisfireThreshold(long misfireThreshold) {
		if (misfireThreshold < 1) {
			throw new IllegalArgumentException("Misfirethreashold must be larger than 0");
		}
		this.misfireThreshold = misfireThreshold;
	}

	@Override
	public void storeTrigger(Trigger newTrigger, boolean replaceExisting) throws JobPersistenceException {
		TriggerWrapper tw = new TriggerWrapper(newTrigger.clone());
		synchronized (lock) {
			if (triggersByKey.get(tw.getTrigger().getKey()) != null) {
				if (!replaceExisting) {
					throw new ObjectAlreadyExistsException(newTrigger);
				}
				removeTrigger(newTrigger.getKey(), false);
			}
			//			if (retrieveJob(newTrigger.getJobKey()) == null) {
			//				throw new JobPersistenceException("The job (" + newTrigger.getJobKey() + ") referenced by the trigger does not exist.");
			//			}
			// add to triggers array
			triggers.add(tw);
			// add to triggers by group
			HashMap<TriggerKey, TriggerWrapper> grpMap = triggersByGroup.get(newTrigger.getKey().getGroup());
			if (grpMap == null) {
				grpMap = new HashMap<TriggerKey, TriggerWrapper>(100);
				triggersByGroup.put(newTrigger.getKey().getGroup(), grpMap);
			}
			grpMap.put(newTrigger.getKey(), tw);
			// add to triggers by FQN map
			triggersByKey.put(tw.getTrigger().getKey(), tw);
			//
			if (pausedTriggerGroups.contains(newTrigger.getKey().getGroup())) {
				//if (pausedTriggerGroups.contains(newTrigger.getKey().getGroup()) || pausedJobGroups.contains(newTrigger.getJobKey().getGroup())) {
				tw.state = TriggerWrapper.STATE_PAUSED;
				if (blockedJobs.contains(tw.getTrigger().getJobKey())) {
					tw.state = TriggerWrapper.STATE_PAUSED_BLOCKED;
				}
			} else if (blockedJobs.contains(tw.getTrigger().getJobKey())) {
				tw.state = TriggerWrapper.STATE_BLOCKED;
			} else {
				timeTriggers.add(tw);
			}
		}
	}

	/*
	 * (non-Javadoc)
	 * @see org.go.core.WorkerStore#supportsPersistence()
	 */
	@Override
	public boolean supportsPersistence() {
		return false;
	}

	@Override
	public void triggeredJobComplete(Trigger trigger, Object jobDetail, CompletedExecutionInstruction setAllJobTriggersError) {
		synchronized (lock) {

		}
	}

	/*
	 * (non-Javadoc)
	 * @see org.go.core.WorkerStore#triggersFired(java.util.List)
	 */
	@Override
	public List<TriggerFiredResult> triggersFired(List<Trigger> triggerss) {
		synchronized (lock) {
			List<TriggerFiredResult> results = new ArrayList<TriggerFiredResult>();
			for (Trigger trigger : triggerss) {
				TriggerWrapper tw = triggersByKey.get(trigger.getTriggerKey());
				// was the trigger deleted since being acquired?
				if (tw == null || tw.trigger == null) {
					continue;
				}
				// was the trigger completed, paused, blocked, etc. since being acquired?
				if (tw.state != TriggerWrapper.STATE_ACQUIRED) {
					continue;
				}
				//
				Calendar cal = null;
				if (tw.trigger.getCalendarName() != null) {
					cal = retrieveCalendar(tw.trigger.getCalendarName());
					if (cal == null)
						continue;
				}
				Date prevFireTime = trigger.getPreviousFireTime();
				// in case trigger was replaced between acquiring and firing
				timeTriggers.remove(tw);
				// call triggered on our copy, and the scheduler's copy
				tw.trigger.triggered(cal);
				trigger.triggered(cal);
				//tw.state = TriggerWrapper.STATE_EXECUTING;
				tw.state = TriggerWrapper.STATE_WAITING;
				TriggerFiredBundle bndle = new TriggerFiredBundle(retrieveJob(null), trigger, cal, false, new Date(), trigger.getPreviousFireTime(), prevFireTime, trigger.getNextFireTime());
				results.add(new TriggerFiredResult(bndle));
			}
			return results;
		}
	}
}
